前情提要

上一篇文章實作了 Qdrant 向量資料庫，因為他免費加上可以本地假設的性質讓我選擇使用他。那今天就要來將 Embedding Model、Retriever、Qdrant、LLM 給結合起來，來讓 AI 可以透過 Retriever 的資料來回答他原本不知道的問題。

檢索增強生成 Retrieval-Augmented Generation (RAG)

簡單來說，透過檢索資料生成回覆，這就是 RAG 的意思。上面這個圖很好的說明 RAG 的流程 (取自 LangChain)：使用者輸入一個問題 -> 進入資料庫檢索 -> 將檢索的結果放入 Prompt 中 -> 將 Prompt 傳遞給 AI -> AI 生成結果，可以想像成向量資料庫是 AI 的 ChatGPT。

Documents Loader & Text Splitter

在實作之前，先再來了解兩個工具：

Documents Loader

因為我們資料來源有很多種檔案可能，但是他可以幾乎可以解析當前最熱門的所有檔案格式，像是 PDF、TXT、Markdown、CSV、HTML、JSON。

Text Splitter

由 Documents Loader 解析成字串之後，要藉由 TextSplitter 將內容隨機切割，若內容太長的話在可能丟給 AI 會超過 token 上限導致錯失重要內容，或者因為太多贅字而檢索到錯誤的內容。等等因素都證明 TextSplitter 是為了更精細地控制檢索和生成過程，從而提升 RAG 系統的準確性、效率和整體性能。

PDF 實戰🔥

PDF 檔案連結 -> 高雄凱米颱風受災慰問補助說明.pdf

from langchain_community.document_loaders import PyPDFLoader
from langchain_text_splitters import RecursiveCharacterTextSplitter

# 讀取 pdf
loader = PyPDFLoader("高雄凱米颱風受災慰問補助說明.pdf")

# 解析 pdf
docs = loader.load()

# 設定切割的長度
text_splitter = RecursiveCharacterTextSplitter(chunk_size=100, chunk_overlap=20)

# 切割內容
splits = text_splitter.split_documents(docs)

# 查看切割後的結果
print(splits)

程式碼結果探討 🧐：

• 附上 PDF 下載連結～有興趣實作的可以取用
• 上圖可以看到已經將內容以我設定的 100 個字的 chunk_size，即代表 100 個字切割一次。然後有可能在重要的地方被切斷，所以 chunk_overlap 的設置會向前覆蓋，那我這邊是向前覆蓋 20 個字。根據我自己的經驗是大約取切割字數的 10% 上下差不多。

HTML 實戰🔥

from langchain_community.document_loaders import WebBaseLoader
from langchain_text_splitters import RecursiveCharacterTextSplitter

# 讀取 Web
loader = WebBaseLoader("https://www.businessweekly.com.tw/international/blog/3016080")

# 解析 Web
docs = loader.load()

# 設定切割的長度
text_splitter = RecursiveCharacterTextSplitter(chunk_size=500, chunk_overlap=50)

# 切割內容
splits = text_splitter.split_documents(docs)

# 查看結果
print(splits)

程式碼結果探討 🧐：

• 內容太多就不截圖了，抓網頁會發生一種可能是可能有廣告會影響資料的乾淨程度，所以更凸顯切割內容的重要性！
• 基本上就跟 PDF 差不多，就是換一個解析的套件即可，各種檔案的類型幾乎都可以如法炮製！

RAG 實戰🔥

這邊所使用的資料就是前面的 PDF 檔案和 HTML 網頁，向量資料庫為 Qdrant，LLM 為台智雲的 ffm-llama3-70b-chat，Embedding Model 也是台智雲的。

套件版本

python version -> 3.11.7
langchain==0.2.14
langchain-core==0.2.33
langchain-community==0.2.12
langchain-qdrant==0.1.3
langchain-text-splitters==0.2.2
qdrant-client=1.11.0
python-dotenv==1.0.1
bs4==0.0.2
pypdf==4.3.1
requests==2.32.3

程式碼

# 匯入套件
from langchain_ffm import ChatFormosaFoundationModel, FFMEmbedding
from langchain_qdrant import QdrantVectorStore
from langchain_community.document_loaders import WebBaseLoader
from langchain_community.document_loaders import PyPDFLoader
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_core.runnables import RunnablePassthrough, RunnableLambda, RunnableParallel
from langchain_core.prompts import PromptTemplate
from qdrant_client import QdrantClient
from qdrant_client.http import models
from qdrant_client.http.models import Distance

# 選擇模型和連線向量資料庫
llm = ChatFormosaFoundationModel(model="ffm-llama3-70b-chat", temperature=0.01)
embeddings = FFMEmbedding(model='ffm-embedding')
client = QdrantClient(url="http://localhost:6333")
collections_name = "rag_practice"

# 1.若 collection 存在則刪除 (現在沒有recreate函數)
if client.collection_exists(collection_name=collections_name):
	client.delete_collection(collection_name=collections_name)
else:
	pass

# 1.建立一個新的 collections
client.create_collection(
	collection_name=collections_name,
		vectors_config=
			models.VectorParams(
				size=1536,
				distance=Distance.COSINE,
	),
)

# langchain 連線 Qdrant 已存在的 collections
qdrant_vector_store = QdrantVectorStore.from_existing_collection(
	url="http://localhost:6333",
	collection_name=collections_name,
	embedding=embeddings,
)

# 2.切割 html 內容的函數
def split_html(url, html_chunk_size, html_chunk_overlap):
	html_loader = WebBaseLoader(url)
	html_docs = html_loader.load()
	html_text_splitter = RecursiveCharacterTextSplitter(chunk_size=html_chunk_size, chunk_overlap=html_chunk_overlap)
	html_split = html_text_splitter.split_documents(html_docs)
	return html_split

# 2.切割 pdf 內容的函數
def split_pdf(pdf_path, pdf_chunk_size, pdf_chunk_overlap):
	pdf_loader = PyPDFLoader(pdf_path)
	pdf_docs = pdf_loader.load()
	pdf_text_splitter = RecursiveCharacterTextSplitter(chunk_size=pdf_chunk_size, chunk_overlap=pdf_chunk_overlap)
	pdf_split = pdf_text_splitter.split_documents(pdf_docs)
	return pdf_split

# 2.將切割的兩個函數透過平行運算的方式進行切割，速度會快一點
def split_parallel():
	process = RunnableParallel({
		"html": RunnableLambda(lambda inputs: split_html(inputs["url"], inputs["html_chunk_size"], inputs["html_chunk_overlap"])),
		"pdf": RunnableLambda(lambda inputs: split_pdf(inputs["pdf_path"], inputs["pdf_chunk_size"], inputs["pdf_chunk_overlap"]))
	})
	results = process.batch([{
		"url": "https://www.businessweekly.com.tw/international/blog/3016080",
		"html_chunk_size": 1000,
		"html_chunk_overlap": 100,
		"pdf_path": "高雄凱米颱風受災慰問補助說明.pdf",
		"pdf_chunk_size": 500,
		"pdf_chunk_overlap": 50
	}])
	return results[0]['html'], results[0]['pdf']

# 2.取得切割結果
html_data, pdf_data = split_parallel()

# 3.將資料匯入 Qdrant 的函數
def data2qdrant(data):
	qdrant_vector_store.add_documents(documents=data)

# 3.匯入資料庫一樣以平行運算的方式匯入
process = RunnableParallel({
	"html_qdrant_vector_store": RunnableLambda(lambda inputs: data2qdrant(inputs["html_data"])),
	"pdf_qdrant_vector_stroe": RunnableLambda(lambda inputs: data2qdrant(inputs["pdf_data"]))
})
process.batch([{"html_data": html_data, "pdf_data": pdf_data}])

# 4.指定 retriever
retriever = qdrant_vector_store.as_retriever()

# 4.RAG 的 prompt
template = """僅根據以下內容回答問題:
{context}
問題: {question}
"""
prompt = PromptTemplate.from_template(template)

# 4.將所有東西都 Chain 起來
chain = (
	RunnableParallel({"context": retriever, "question": RunnablePassthrough()})
	| prompt
	| llm
)

# 5.透過 batch 方式一次傳多個問題取得想要的資訊
q1_for_olympics = "奧運男籃第一隊的球員有誰？"
q2_for_olympics = "奧運男籃的A,B,C組分別是哪些國家？"
q_for_typhoon = "淹水50公分以上住戶可以領多少錢？"
response = chain.batch([q1_for_olympics, q_for_typhoon, q2_for_olympics])
print(f"""Q: {q1_for_olympics}, \nA: {response[0].content}\n
Q: {q2_for_olympics}, \nA: {response[2].content}\n
Q: {q_for_typhoon}, \nA: {response[1].content}""")

程式碼結果探討 🧐 (與註解編號一致)：

1. 這邊要先刪除再建立的原因是如果 collections 存在的話，沒有刪掉的話後續再寫入資料會一直往後加，這樣就會有很多重複的資料，所以才會做這個判斷，然後固定新建一個 collections。
2. 切割這邊因為網頁多餘的資訊比較多，所以我 chunk_size 設比較大，pdf 就設比較小。然後因為切割完一個資料再切割另一個資料，如果資料量多的話會很久，所以使用 LangChain LCEL 的架構的平行運算函數，可以同時進行切割。
3. 資料匯入 Qdrant 跟前面也是一樣的道理，同時匯入比較省時，資料量大的話會有明顯差異。
4. 最後就是將 Retriever, Prompt, LLM 給串在一起，Retriever 那邊的意思是，使用者的 query 就是 question，所以設定 RunnablePassthrough，然後這個 query 同時也會傳去 Retriever，然後會回傳檢索的結果傳進 retriever 這個變數，這是 LCEL 的精髓所在。
5. 最後就是透過 batch 方式問多個問題，可以看到 AI 回答的內容都是精準且正確的！

結論

今天我們成功實作了 RAG 的功能，也成功使用到 LCEL 架構，也運用平行運算加速整個過程，也成功連結資料庫並檢索到正確的結果讓 AI 回答正確。透過這個實作更熟悉 LangChain 框架，還有 RAG 技術的應用，還蠻有成就感的☺️

題外話🤣

Midjourney 有網頁版了！！

下一篇文章：讓 AI 記得你 - Memory 對話功能